package com.dbms.execution.execution;

import com.dbms.catalog.Catalog;
import com.dbms.common.Column;
import com.dbms.common.Schema;
import com.dbms.common.Tuple;
import com.dbms.common.Type;
import com.dbms.execution.plan.PlanNode;
import com.dbms.execution.udf.FunctionRegistry;
import com.dbms.execution.udf.Udf;
import com.dbms.parser.ast.FunctionCallNode;
import com.dbms.storage.BufferPoolManager;
import com.dbms.storage.DiskManager;
import com.dbms.storage.Page;
import com.dbms.storage.TablePage;

import java.io.File;
import java.io.IOException;
import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.stream.Collectors;

/**
 * ProjectionExecutor (投影执行器) 负责处理 SQL 查询的 SELECT 子句。
 * "投影" 在关系代数中意味着选择某些列。这个执行器将从其子执行器传来的
 * "宽" 元组转换为最终用户需要的 "窄" 元组。
 * 它支持：
 * - 选择特定的列 (e.g., `SELECT id, name`)
 * - 选择所有列 (e.g., `SELECT *`)
 * - 在 SELECT 子句中使用函数 (e.g., `SELECT UPPER(name), id + 100`)
 */
public class ProjectionExecutor implements Executor {
    private final Executor child; // 子执行器，提供原始的、完整的元组
    private final Schema outputSchema; // 投影操作后最终输出的元组模式
    /**
     * 投影项列表，可以包含 String (列名) 或 FunctionCallNode (函数调用)。
     */
    private final List<Object> projectionItems;
    private final FunctionRegistry functionRegistry; // 用于查找和执行 UDF

    // 用于 EXPLAIN ANALYZE 的性能指标
    private long startTimeNanos;
    private long totalTimeNanos;
    private long rowsProduced;

    public ProjectionExecutor(Schema outputSchema, List<Object> projectionItems, Executor child, FunctionRegistry functionRegistry) {
        this.outputSchema = outputSchema;
        this.projectionItems = projectionItems;
        this.child = child;
        this.functionRegistry = functionRegistry;
    }

    @Override
    public void open() {
        this.rowsProduced = 0;
        this.startTimeNanos = System.nanoTime();
        child.open();
    }

    @Override
    public Tuple next() {
        // 1. 从子执行器获取一个原始元组
        Tuple inputTuple = child.next();
        if (inputTuple == null) {
            return null; // 子执行器已无数据
        }

        List<Object> outputValues = new ArrayList<>();
        Schema inputSchema = child.getSchema();

        // 2. 遍历 SELECT 列表中的每一项（列名或函数）
        for (Object item : projectionItems) {
            if (item instanceof String) {
                // 如果是简单的列名或 "*"，则调用字符串处理器
                handleStringProjection((String) item, inputTuple, inputSchema, outputValues);
            } else if (item instanceof FunctionCallNode) {
                // 如果是函数调用，则调用函数处理器
                handleFunctionProjection((FunctionCallNode) item, inputTuple, inputSchema, outputValues);
            }
        }

        this.rowsProduced++;
        // 3. 使用生成的值列表和输出模式创建新的、经过投影的元组
        return new Tuple(outputValues, this.outputSchema);
    }

    /**
     * 处理简单的列名投影 (e.g., "id", "name", "*")。
     */
    private void handleStringProjection(String columnName, Tuple inputTuple, Schema inputSchema, List<Object> outputValues) {
        // 处理 SELECT * 的情况
        if ("*".equals(columnName)) {
            for (int i = 0; i < inputSchema.getColumnCount(); i++) {
                outputValues.add(inputTuple.getValue(i));
            }
            return;
        }

        // 处理单个列名的情况
        int inputIndex = findColumnIndex(inputSchema, columnName);
        if (inputIndex != -1) {
            outputValues.add(inputTuple.getValue(inputIndex));
        } else {
            // 这通常应该在语义分析阶段被捕获，但作为一道安全防线
            throw new RuntimeException("Execution Error: Column '" + columnName + "' not found in input tuple.");
        }
    }

    /**
     * 处理函数调用投影 (e.g., "UPPER(name)")。
     */
    private void handleFunctionProjection(FunctionCallNode funcNode, Tuple inputTuple, Schema inputSchema, List<Object> outputValues) {
        // 1. 从函数注册表中查找函数定义
        Optional<Udf> udfOpt = functionRegistry.getFunction(funcNode.getFunctionName());
        if (udfOpt.isEmpty()) {
            throw new RuntimeException("Execution Error: Unknown function '" + funcNode.getFunctionName() + "'");
        }
        Udf udf = udfOpt.get();

        // 2. 准备函数的参数列表
        List<Object> args = new ArrayList<>();
        for (Object argDesc : funcNode.getArguments()) {
            if (argDesc instanceof String) {
                // 参数可能是列名（需要从元组取值）或字符串字面量
                int colIndex = findColumnIndex(inputSchema, (String) argDesc);
                if (colIndex != -1) {
                    args.add(inputTuple.getValue(colIndex)); // 从元组中获取列值作为参数
                } else {
                    args.add(argDesc); // 如果在输入模式中找不到，则认为是字符串字面量
                }
            } else {
                args.add(argDesc); // 认为是数字或其他类型的字面量
            }
        }

        // 3. 执行函数并将结果添加到输出值列表中
        try {
            Object result = udf.execute(args.toArray());
            outputValues.add(result);
        } catch (Exception e) {
            throw new RuntimeException("Execution Error: Failed to execute function '" + funcNode.getFunctionName() + "'", e);
        }
    }

    /**
     * 辅助方法：根据列名在其 Schema 中查找索引。
     */
    private int findColumnIndex(Schema schema, String columnName) {
        for (int i = 0; i < schema.getColumnCount(); i++) {
            if (schema.getColumn(i).getColumnName().equalsIgnoreCase(columnName)) {
                return i;
            }
        }
        return -1; // 未找到
    }

    @Override
    public void close() {
        child.close();
        this.totalTimeNanos = System.nanoTime() - startTimeNanos;
    }

    @Override
    public Schema getSchema() {
        return this.outputSchema;
    }

    @Override
    public PlanNode getPlanNode() {
        String columns = projectionItems.stream()
                .map(Object::toString)
                .collect(Collectors.joining(", "));
        PlanNode node = new PlanNode("Projection", "output: [" + columns + "]");
        node.setExecutionTimeNanos(this.totalTimeNanos);
        node.setRowsProduced(this.rowsProduced);
        node.addChild(child.getPlanNode());
        return node;
    }
}